#include "TcpConnection.h"
#include "Socket.h"
#include "Channel.h"
#include <functional>
#include "Logger.h"
#include <memory>
#include "EventLoop.h"
namespace mymuduo {

static EventLoop* CheckLoopNotNull(EventLoop* loop) {
    if(loop == nullptr) {
        LOG_FATAL("%s:%s:%d connection loop is null!\n", __FILE__, __FUNCTION__, __LINE__);
    }
    return loop;
}

TcpConnection::TcpConnection(EventLoop* loop,
    const std::string& name,
    int sockfd,
    const InetAddress& localAddr,
    const InetAddress& peerAddr)
    : loop_(CheckLoopNotNull(loop)),
    name_(name),
    state_(kConnecting),
    reading_(true),
    socket_(new Socket(sockfd)),
    channel_(new Channel(loop, sockfd)),
    localAddr_(localAddr),
    peerAddr_(peerAddr),
    highWaterMark_(64*1024*1024)
{
    channel_->setReadCallback(
    std::bind(&TcpConnection::handleRead, this, std::placeholders::_1));
    channel_->setWriteCallback(
    std::bind(&TcpConnection::handleWrite, this));
    channel_->setCloseCallback(
    std::bind(&TcpConnection::handleClose, this));
    channel_->setErrorCallback(
    std::bind(&TcpConnection::handleError, this));
    LOG_DEBUG ("TcpConnection::ctor[ %s ] at fd=%d\n", name_, sockfd);
    socket_->setKeepAlive(true);
}

TcpConnection::~TcpConnection()
{
  LOG_DEBUG("TcpConnection::dtor[ %s ] at fd=%d state=%d\n", name_, channel_->fd(), (int)state_);
}

void TcpConnection::handleRead(TimeStamp receiveTime) {
    int savedErrno = errno;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    if(n > 0) {
        // 已建立连接的用户，有可读事件发生了，调用用户传递的回调操作
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    } else if(n == 0) {
        handleClose();
    } else {
        errno = savedErrno;
        LOG_ERROR("TcpConnection::handleRead() error\n");
        handleError();
    }
}

void TcpConnection::send(const std::string& buf) {
    if(state_ == kConnected) {
        if(loop_->isInLoopThread()) {
            sendInLoop(buf.c_str(), buf.size());
        } else {
            loop_->runInLoop(std::bind(&TcpConnection::sendInLoop, this, buf.c_str(), buf.size()));
        }
    }
}

// 应用写的慢，内核发送数据慢，需要把待发送数据写入缓冲区，并设置高水位回调
void TcpConnection::sendInLoop(const void* message, size_t len) {
    ssize_t nwrote = 0;
    ssize_t remaining = len;
    bool faultErrno = false;
    // 之前调用过该 connection 的 shutdown，不再发送
    if(state_ == kDisconnected) {
        LOG_ERROR("disconnected, give up writing!\n");
        return;
    }
    // 表示 channel_ 第一次开始写数据，而且缓冲区没有带发送数据
    if(!channel_->isWriting() && outputBuffer_.readableBytes() == 0) {
        nwrote = ::write(channel_->fd(), message, len);
        if(nwrote > 0) {
            remaining = len - nwrote;
            if(remaining == 0 && writeCompleteCallback_) {
                // 数据全部发送完成，不用给 channel设置 epollout 事件
                loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
            }
        } else {
            // 出错
            nwrote = 0;
            if(errno != EWOULDBLOCK) {
                LOG_ERROR("TcpConnection::sendInLoop() error!\n");
                if(errno == EPIPE || errno == ECONNRESET) {
                    faultErrno = true;
                }
            }
        }
    }
    if(!faultErrno && remaining > 0) {
        // 当前这一次 write 并没有把数据全部发送出去，剩余的数据需要保存在缓冲区，然后给 channel 注册 epollout 事件
        // 注册 epollout 事件，poller 发现 tcp 的发送缓冲区有空间，会通知相应的 sock-channel，调用 handleWrite 回调方法
        // 最终也就是调用 TcpConnection:handleWrite 方法，把发送缓冲区中的数据全部发送完成
        size_t oldLen = outputBuffer_.readableBytes(); // 目前发送缓冲区剩余待发送数据的长度
        if(oldLen + remaining >= highWaterMark_ && oldLen < highWaterMark_ && highWaterMark_) {
            loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
        }
        outputBuffer_.append((char*)message + nwrote, remaining);
        if(!channel_->isWriting()) {
            channel_->enableWriting(); // 这里一定要注册 channel 的写事件，否则 poller 不会给 channel 通知 epollout
        }
    }
}

void TcpConnection::shutdown() {
    if(state_ == kConnected) {
        setState(kDisconnecting);
        loop_->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
    }
}

void TcpConnection::shutdownInLoop() {
    if(!channel_->isWriting()) {
        // 说明当前 outputBuffer 中的数据已经全部发送完成
        socket_->shutdownWrite(); // 关闭写端
    }
}

void TcpConnection::handleWrite() {
    if(channel_->isWriting()) {
        int savedErrno = 0;
        ssize_t n = outputBuffer_.writeFd(channel_->fd(), &savedErrno);
        if(n > 0) {
            outputBuffer_.retrieve(n);
            if(outputBuffer_.readableBytes() == 0) {
                channel_->disableWriting();
                if(writeCompleteCallback_) {
                    // 唤醒 loop_ 对应的线程，执行回调
                    loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
                }
                if(state_ == kDisconnected) {
                    shutdownInLoop();
                }
            }
        } else {
            LOG_ERROR("TcpConnection::handleWrite() errno\n");
        }
    } else {
        LOG_ERROR("TcpConnection fd=%d is down, no more writing\n", channel_->fd());
    }
}

// poller => channel::closeCallback => TcpConnection::handleClose
void TcpConnection::handleClose() {
    LOG_INFO("fd=%d state=%d\n", channel_->fd(), (int)state_);
    setState(kDisconnected);
    channel_->disableAll();
    TcpConnectionPtr connptr = (shared_from_this());
    connectionCallback_(connptr); // 执行连接关闭的回调
    closeCallback_(connptr); // 关闭连接的回调 执行的是 TcpServer::removeConnection
}

void TcpConnection::handleError() {
    int optval, err;
    socklen_t optlen = sizeof(optval);
    if(::getsockopt(channel_->fd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0) {
        err = errno;
    } else {
        err = optval;
    }
    LOG_ERROR("TcpConnection::handleError() name:%s - SO_ERROR%d\n", name_.c_str(), err);
}


void TcpConnection::connectEstablished() {
    setState(kConnected);
    channel_->tie(shared_from_this());
    channel_->enableReading();
    connectionCallback_(shared_from_this());
} 

void TcpConnection::connectDestroyed() {
    if(state_ == kConnected) {
        setState(kDisconnected);
        channel_->disableAll(); // 把 channel 所有感兴趣的事件，从 poller 中 del 掉
        connectionCallback_(shared_from_this());
    }
    channel_->remove(); // channel 从 poller 中删除掉
}

}